package xjh.him.netty;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInboundHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import xjh.him.netty.codec.LengthFieldDecoder;
import xjh.him.netty.codec.MsgDecoder;
import xjh.him.netty.codec.MsgEncoder;
import xjh.him.netty.codec.Packet;
import xjh.him.netty.dto.HeartBeat;
import xjh.him.netty.handler.BaseHandler;
import xjh.him.util.LogcatUtil;

public class NettyClient {
    private static boolean isUnixPlatform = false;
    private EventLoopGroup workerGroup;
    private Bootstrap bootstrap;
    private List<ChannelInboundHandler> handlerList;
    private ChannelFuture future;

    static {
        if (System.getProperty("os.name").toLowerCase().contains("linux")) {
            isUnixPlatform = true;
        }
    }

    private boolean useEpoll() {
        return isUnixPlatform && Epoll.isAvailable();
    }

    public NettyClient() {
        if (useEpoll()) {
            workerGroup = new EpollEventLoopGroup(3, new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("NettyEpollWorker_%d", threadIndex.getAndIncrement()));
                }
            });
        }else {
            workerGroup = new NioEventLoopGroup(3, new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("NettyNioWorker_%d", threadIndex.getAndIncrement()));
                }
            });
        }
        initHandlerList();
        initBootstrap();
    }

    private void initHandlerList() {
        handlerList = new ArrayList<>();
        handlerList.add(new BaseHandler());
    }

    private void initBootstrap() {
        bootstrap = new Bootstrap();
        bootstrap.group(workerGroup)
            .channel(useEpoll() ? EpollSocketChannel.class :
                    NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_KEEPALIVE, false)
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
            .option(ChannelOption.SO_SNDBUF, 65535)
            .option(ChannelOption.SO_RCVBUF, 65535)
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel channel) {
                    ChannelPipeline pipeline = channel.pipeline();

                    pipeline.addLast("Client LengthFieldDecoder", new LengthFieldDecoder());
                    pipeline.addLast("Client Decoder", new MsgDecoder());
                    for (ChannelInboundHandler handler : handlerList) {
                        pipeline.addLast(handler);
                    }
                    pipeline.addLast("Client Encoder", new MsgEncoder());
                }
            });
    }

    public void connect(InetSocketAddress address) {
        future = bootstrap.connect(address);
        future.addListener((ChannelFutureListener) f -> {
            if (f.isSuccess()) {
                LogcatUtil.i("client connect %s success.", address);
                //startHeartBeatThread();
            }else {
                f.channel().pipeline().fireChannelInactive();
                LogcatUtil.i("client connect %s failed: %s", address, f.cause().getMessage());
            }
        });
    }

    private void startHeartBeatThread() {
        Thread heartBeat = new Thread(() -> {
            future.channel().eventLoop().scheduleAtFixedRate(() -> {
                LogcatUtil.d("send heart beat msg");
                sendMsg(Packet.create(HeartBeat.class, new HeartBeat()));
            }, Constant.IDLE_READ_TIMEOUT, Constant.IDLE_READ_TIMEOUT, TimeUnit.SECONDS);
        }, "HeartBeat");
        heartBeat.start();
        LogcatUtil.i("heartBeatThread start");
    }

    public ChannelFuture sendMsg(Object msg) {
        return future.channel().writeAndFlush(msg);
    }

    public void shutdownNow() {
        future.channel().close().addListener(f -> {
            if (f.isSuccess()) {
                LogcatUtil.i("Netty channel close success");
            }
        });
    }
}
